package io.github.junxworks.junx.stream.pipeline;

import io.github.junxworks.junx.core.executor.StandardThreadExecutor;
import io.github.junxworks.junx.core.util.ObjectUtils;
import io.github.junxworks.junx.event.impl.ExecutorEventChannel;
import lombok.Getter;
import lombok.Setter;

import java.util.concurrent.ThreadPoolExecutor;

/**
 * 管道中的基础组件
 *
 * @author 王兴
 * @date 2024/09/10
 */
@Getter
@Setter
public class ConcurrentComponent extends BaseComponent {

    /**
     * @return
     */
    @Override
    public boolean isDone() {
        if (!super.isDone()) {
            return false;
        }
        StandardThreadExecutor executorService = (StandardThreadExecutor) ((ExecutorEventChannel) eventChannel).getExecutor();
        ThreadPoolExecutor threadPoolExecutor = (ThreadPoolExecutor) ObjectUtils.mirror().on(executorService).get().field("executor");
        return threadPoolExecutor.getActiveCount() == 0 && threadPoolExecutor.getQueue().isEmpty();
    }

    /**
     * @throws Exception
     */
    @Override
    public void initialize() throws Exception {
        eventChannel = new ExecutorEventChannel(stepName, inputTopic);
        eventChannel.setEventHandler(eventChannelHandler);
        //如果队列满了，让调用者分担压力执行任务
        ((ExecutorEventChannel) eventChannel).setCallerRunTaskWhenBusy(true);
        ((ExecutorEventChannel) eventChannel).setBufferSize(this.bufferSize);
    }

    /**
     * @throws Exception
     */
    @Override
    public void destroy() throws Exception {

    }
}
